package com.edu.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/20.
 *
 * @author Topus
 * @desc 考试域：考试成绩明细
 */
public class DwdTestExam {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_test_exam_group"));
        Table resultTb = tableEnv.sqlQuery("select data['id'] id, data['paper_id'] paper_id, data['user_id'] user_id,\n"
            + "data['score'] score, data['duration_sec'] duration_sec, data['deleted'] deleted, ts\n"
            + "from topic_db\n" + "where `table`='test_exam' and `type`='insert'");
        tableEnv.createTemporaryView("result_table", resultTb);

        tableEnv.executeSql("create table dwd_test_exam(\n" + "id string,\n" + "paper_id string,\n"
            + "user_id string,\n" + "score string,\n" + "duration_sec string,\n" + "deleted string,\n" + "ts string,\n"
            + "primary key(id) not enforced\n" + ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_exam"));

        tableEnv.executeSql("insert into dwd_test_exam select * from result_table");
    }
}
